package com.zebra.remoting.connection;

import com.zebra.common.concurrent.ZebraThreadFactory;
import com.zebra.remoting.CommandHandler;
import com.zebra.remoting.RemotingHandler;
import com.zebra.remoting.RpcCodec;
import com.zebra.remoting.config.ConfigManager;
import com.zebra.remoting.config.ServerType;
import com.zebra.remoting.util.NettyEventLoopUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.NonNull;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * ConnectionFactory to create connection.
 * @author sk (415937387@qq.com)
 */
public class DefaultConnectionFactory implements ConnectionFactory {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultConnectionFactory.class);

    private static final EventLoopGroup workerGroup = NettyEventLoopUtil.newEventLoopGroup(Runtime
                                                        .getRuntime().availableProcessors() + 1,
                                                        new ZebraThreadFactory(
                                                            "zebra-client-worker", true));

    protected  Bootstrap bootstrap;

    private   RpcCodec codec;
    private   CommandHandler commandHandler;
    private   ChannelHandler heartbeatHandler;

    @Override
    public void registerRpcCodec(@NonNull RpcCodec rpcCodec) {
        this.codec = rpcCodec;
    }

    @Override
    public void registerCommandHandler(@NonNull CommandHandler commandHandler) {
        this.commandHandler = commandHandler;
    }

    @Override
    public void registerHeartbeatProcessor(@NonNull ChannelHandler processor) {
        this.heartbeatHandler = processor;
    }


    @Override
    public void init() {
        bootstrap = new Bootstrap();
        bootstrap.group(workerGroup).channel(NettyEventLoopUtil.getClientSocketChannelClass())
            .option(ChannelOption.SO_SNDBUF, ConfigManager.tcp_so_sndbuf())
            .option(ChannelOption.SO_RCVBUF, ConfigManager.tcp_so_rcvbuf());

        // init netty write buffer water mark
        initWriteBufferWaterMark();

//        final boolean flushConsolidationSwitch = this.confInstance.switches().isOn(
//            GlobalSwitch.CODEC_FLUSH_CONSOLIDATION);
        final RemotingHandler rpcHandler = new RemotingHandler(commandHandler);

        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) {
                ChannelPipeline pipeline = channel.pipeline();
                pipeline.addLast(codec.newLengthDecoder());

                pipeline.addLast("heartbeatHandler", heartbeatHandler);
                pipeline.addLast("handler", rpcHandler);
            }
        });
    }

    /**
     * init netty write buffer water mark
     */
    private void initWriteBufferWaterMark() {
        int lowWaterMark = ConfigManager.netty_buffer_low_watermark();
        int highWaterMark = ConfigManager.netty_buffer_high_watermark();
        if (lowWaterMark > highWaterMark) {
            throw new IllegalArgumentException(
                String
                    .format(
                        "[client side] zebra netty high water mark {%s} should not be smaller than low water mark {%s} bytes)",
                        highWaterMark, lowWaterMark));
        } else {
            logger.warn(
                "[client side] zebra netty low water mark is {} bytes, high water mark is {} bytes",
                lowWaterMark, highWaterMark);
        }
        this.bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
            lowWaterMark, highWaterMark));
    }

    public Connection connection(String ip, int port, int serverId, ServerType severType, int timeOut)throws Exception{
        Channel channel = connect(ip, port, timeOut);
        return new Connection(channel, severType, serverId);
    }


    public Channel connect(String targetIP, int targetPort, int connectTimeout)throws Exception {
        // prevent unreasonable value, at least 1000
        connectTimeout = Math.max(connectTimeout, 1000);
        String address = targetIP + ":" + targetPort;
        if (logger.isDebugEnabled()) {
            logger.debug("connectTimeout of address [{}] is [{}].", address, connectTimeout);
        }
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(targetIP, targetPort));

        future.awaitUninterruptibly();
        if (!future.isDone()) {
            String errMsg = "Create connection to " + address + " timeout!";
            logger.error(errMsg);
            throw new Exception(errMsg);
        }
        if (future.isCancelled()) {
            String errMsg = "Create connection to " + address + " cancelled by user!";
            logger.error(errMsg);
            throw new Exception(errMsg);
        }
        if (!future.isSuccess()) {
            String errMsg = "Create connection to " + address + " error!";
            logger.error(errMsg);
            throw new Exception(errMsg, future.cause());
        }
        return future.channel();
    }


}
